package com.mspbots.cw.amqp.receivers;

import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.node.ArrayNode;
import com.fasterxml.jackson.databind.node.ObjectNode;
import com.mspbots.cw.amqp.BaseReceiver;
import com.mspbots.cw.amqp.CallbackAction;
import com.mspbots.cw.amqp.model.*;
import lombok.RequiredArgsConstructor;
import lombok.extern.log4j.Log4j2;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.data.relational.core.query.Criteria;
import org.springframework.stereotype.Component;
import org.springframework.util.ObjectUtils;
import reactor.core.publisher.Mono;

import java.time.Duration;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.Optional;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Function;

/**
 * com.mspbots.core.amqp.receivers.TicketsReceiver
 *
 * @author Alex bob(https://github.com/vnobo)
 * @date Created by 2019/8/12
 */
@Log4j2
@Component
@RequiredArgsConstructor
public class TicketsReceiver extends BaseReceiver {

    @RabbitListener(queues = "core.tickets")
    public void processMessage(Ticket ticket) {
        try{
            log.debug("Receiver tenant [{}] tickets [{}]", ticket.getTenantId(), ticket.getCwId());

            Ticket oldTicket = this.databaseClient.select().from(Ticket.class)
                    .matching(Criteria.where("tenantId").is(ticket.getTenantId())
                            .and("cwId").is(ticket.getCwId())).fetch().one().block();

            ObjectNode audiNode = this.objectMapper.createObjectNode();
            audiNode.put("id", !ObjectUtils.isEmpty(ticket.getStatusId()) ? ticket.getStatusId() : 0);
            audiNode.put("name", !ObjectUtils.isEmpty(ticket.getStatus()) ? ticket.getStatus() : "");
            audiNode.put("updatedBy", !ObjectUtils.isEmpty(ticket.getUpdatedBy()) ? ticket.getUpdatedBy() : "");
            audiNode.put("time", !ObjectUtils.isEmpty(ticket.getUpdatedTime()) ?
                    ticket.getUpdatedTime().format(DateTimeFormatter.ISO_DATE_TIME) :
                    LocalDateTime.now().format(DateTimeFormatter.ISO_DATE_TIME));

            if (!ObjectUtils.isEmpty(oldTicket) && oldTicket.getId() > 0) {
                log.debug("Receiver connect wise tickets tenant id:{},cw id:{},ticket update id:{}",
                        ticket.getTenantId() + 963, ticket.getCwId(), oldTicket.getId());
                ArrayNode auditor = ObjectUtils.isEmpty(oldTicket.getAuditor()) || oldTicket.getAuditor().size() == 0 ?
                        this.objectMapper.createArrayNode() : oldTicket.getAuditor().deepCopy();
                if (oldTicket.getStatusId() != ticket.getStatusId()) {
                    auditor.add(audiNode);
                    log.debug("Audi node logs {}", audiNode);
                    setAuditor(ticket, audiNode);

                    TicketAuditTrail auditTrail = TicketAuditTrail.builder()
                            .tenantId(ticket.getTenantId())
                            .ticketId(ticket.getCwId())
                            .statusFrom(oldTicket.getStatus())
                            .statusTo(audiNode.get("name").asText())
                            .identifier(audiNode.get("updatedBy").asText())
                            .updatedTime(LocalDateTime.parse(audiNode.get("time").asText(), DateTimeFormatter.ISO_DATE_TIME))
                            .syncTime(LocalDateTime.now()).build();

                    this.databaseClient.insert().into(TicketAuditTrail.class).using(auditTrail)
                            .fetch().rowsUpdated().block();
                }
                ticket.setAuditor(auditor);
                ticket.setId(oldTicket.getId());
            } else {
                if (ticket.getAction() != CallbackAction.deleted) {
                    ticket.setAuditor(this.objectMapper.createArrayNode().add(audiNode));
                    setAuditor(ticket, audiNode);
                }
            }

            ticket.setSyncTime(LocalDateTime.now());

            var updateMono = this.databaseClient.update().table(Ticket.class).using(ticket).fetch().rowsUpdated();
            var insertMono = this.databaseClient.insert().into(Ticket.class).using(ticket).fetch().rowsUpdated();

            AtomicReference<CallbackAction> atomicReference = new AtomicReference<>(ticket.getAction());
            Function<Mono<Integer>, Mono<String>> filterAndMap = f -> {
                Mono<Integer> source = f.onErrorReturn(0);
                if (atomicReference.get() == CallbackAction.added) {
                    return source.map(val -> "This tickets id [" + ticket.getCwId() + "] is exist, not add.")
                            .doOnNext(log::error);
                }
                return source.map(val -> "This tickets id [" + ticket.getCwId() + "] not exist, not update.")
                        .doOnNext(log::debug);
            };

            if (ticket.getAction() == CallbackAction.added) {
                insertMono.transformDeferred(filterAndMap).block();
            } else if (ticket.getAction() == CallbackAction.updated) {
                updateMono.transformDeferred(filterAndMap).block();
            } else if (ticket.getAction() == CallbackAction.deleted) {
                this.databaseClient.delete().from(Ticket.class)
                        .matching(Criteria.where("tenantId").is(ticket.getTenantId())
                                .and("cwId").is(ticket.getCwId()))
                        .fetch().rowsUpdated().block(Duration.ofSeconds(10));
            } else {
                if (!ObjectUtils.isEmpty(ticket.getId()) && ticket.getId() > 0) {
                    updateMono.block();
                } else {
                    insertMono.block();
                }
            }
        }catch (Exception e){
            this.databaseClient.insert().into(ExceptionLog.class).using(
                    ExceptionLog.builder()
                            .tenantId(ticket.getTenantId())
                            .name("core.tickets")
                            .status(500)
                            .type("core.tickets")
                            .msg(e.getMessage()+","+objectMapper.convertValue(ticket,JsonNode.class).toString() )
                            .build()
            ).fetch().rowsUpdated().block(Duration.ofSeconds(10));
        }

    }

    private void setAuditor(Ticket ticket, ObjectNode audiNode) {
        TicketAuditor ticketAuditor = TicketAuditor.builder()
                .tenantId(ticket.getTenantId())
                .ticketId(ticket.getCwId())
                .statusId(audiNode.get("id").asInt())
                .statusName(audiNode.get("name").asText())
                .updatedBy(audiNode.get("updatedBy").asText())
                .updatedTime(LocalDateTime.parse(audiNode.get("time").asText(), DateTimeFormatter.ISO_DATE_TIME))
                .syncTime(LocalDateTime.now())
                .build();
        this.databaseClient.insert().into(TicketAuditor.class).using(ticketAuditor)
                .fetch().rowsUpdated().block();
    }

    @RabbitListener(queues = "wise.priority")
    public void processPriority(Priority priority) {
        try{
            Mono<Integer> updateMono = this.databaseClient.execute("SELECT id FROM tenant_priorities " +
                    "WHERE tenant_id =:tenantId AND cw_id =:cwId")
                    .as(Long.class)
                    .bind("tenantId", priority.getTenantId())
                    .bind("cwId", priority.getCwId())
                    .fetch().one()
                    .flatMap(id -> {
                        log.debug("Receiver Priority tenant id:{},cw id:{}", priority.getTenantId(), priority.getCwId());
                        priority.setId(id);
                        priority.setSyncTime(LocalDateTime.now());
                        return this.databaseClient.update().table(Priority.class).using(priority).fetch().rowsUpdated();
                    });
            updateMono.switchIfEmpty(this.databaseClient.insert().into(Priority.class).using(priority).fetch().rowsUpdated())
                    .onErrorResume(err -> updateMono)
                    .subscribe(result -> log.debug("Priority save is success {}", result),
                            err -> log.error("Priority save or update is error,msg: {}", err.getMessage()));
//        super.messagingTemplate.convertAndSend("next-ticket.redis.priority",priority);
        }catch (Exception e){
            this.databaseClient.insert().into(ExceptionLog.class).using(
                    ExceptionLog.builder()
                            .tenantId(priority.getTenantId())
                            .name("wise.priority")
                            .status(500)
                            .type("wise.priority")
                            .msg(e.getMessage()+","+objectMapper.convertValue(priority,JsonNode.class).toString() )
                            .build()
            ).fetch().rowsUpdated().block(Duration.ofSeconds(10));
        }
    }

    @RabbitListener(queues = "wise.agreements")
    public void processAgreements(JsonNode agree) {
        try{
            if (ObjectUtils.isEmpty(agree.get("id"))) {
                log.error("agreements is error,msg:{}", agree);
                return;
            }
            Agreements agreements = Agreements.builder()
                    .cwId(agree.get("id").asInt())
                    .companyId(agree.get("company").get("id").asInt())
                    .tenantId(agree.get("tenantId").asInt())
                    .name(agree.get("name").asText())
                    .billAmount(!ObjectUtils.isEmpty(agree.get("billAmount")) ? agree.get("billAmount").asDouble() : 0)
                    .createdTime(!ObjectUtils.isEmpty(agree.get("billStartDate")) ?
                            LocalDateTime.parse(agree.findValue("billStartDate").asText(), DateTimeFormatter.ISO_DATE_TIME) :
                            LocalDateTime.parse(agree.findValue("startDate").asText(), DateTimeFormatter.ISO_DATE_TIME))
                    .updateTime(!ObjectUtils.isEmpty(agree.get("billStartDate")) ?
                            LocalDateTime.parse(agree.findValue("lastUpdated").asText(), DateTimeFormatter.ISO_DATE_TIME) : LocalDateTime.now())
                    .extend(agree)
                    .build();
            this.databaseClient.execute("SELECT id FROM tenant_agreements WHERE tenant_id =:tenantId AND cw_id =:cwId")
                    .as(Long.class)
                    .bind("tenantId", agreements.getTenantId())
                    .bind("cwId", agreements.getCwId())
                    .fetch().one()
                    .flatMap(id -> {
                        log.debug("Receiver agreements tenant id:{},cw id:{}", agree.get("tenantId"), agree.get("id"));
                        agreements.setId(id);
                        agreements.setSyncTime(LocalDateTime.now());
                        return this.databaseClient.update().table(Agreements.class).using(agreements).fetch().rowsUpdated();
                    })
                    .switchIfEmpty(this.databaseClient.insert().into(Agreements.class).using(agreements).fetch().rowsUpdated())
                    .subscribe(result -> log.debug("Agreements save is success {}", result),
                            err -> log.error("Agreements save or update is error,msg: {}", err.getMessage()));
//        ObjectNode node = super.objectMapper.createObjectNode();
//        node.putPOJO("tenantId", agreements.getTenantId());
//        node.putPOJO("companyId", agreements.getCompanyId());
//        super.messagingTemplate.convertAndSend("next-ticket.redis.agreements", node);
        }catch (Exception e){
            if (!ObjectUtils.isEmpty(agree.get("tenantId"))){
                this.databaseClient.insert().into(ExceptionLog.class).using(
                        ExceptionLog.builder()
                                .tenantId(agree.get("tenantId").asInt())
                                .name("wise.agreements")
                                .status(500)
                                .type("wise.agreements")
                                .msg(e.getMessage()+","+agree.toString() )
                                .build()
                ).fetch().rowsUpdated().block(Duration.ofSeconds(10));
            }
        }

    }

    @RabbitListener(queues = "wise.contacts")
    public void processContacts(JsonNode contact) {
        try{
            if (ObjectUtils.isEmpty(contact.get("id"))) {
                log.error("agreements is error,msg:{}", contact);
                return;
            }
            log.debug("Receiver contacts tenant id:{},cw id:{}", contact.get("tenantId"), contact.get("id"));
            String firstName = ObjectUtils.isEmpty(contact.get("firstName")) ? "" : contact.get("firstName").asText();
            String lastName = ObjectUtils.isEmpty(contact.get("lastName")) ? "" : contact.get("lastName").asText();
            Contacts contacts = Contacts.builder()
                    .cwId(contact.get("id").asLong())
                    .tenantId(contact.get("tenantId").asInt())
                    .name(firstName + " " + lastName)
                    .title(ObjectUtils.isEmpty(contact.get("title")) ? null : contact.get("title").asText())
                    .updateTime(LocalDateTime.parse(contact.findValue("lastUpdated").asText(), DateTimeFormatter.ISO_DATE_TIME))
                    .extend(contact)
                    .build();
            Optional<Long> id = this.databaseClient.execute("SELECT id FROM tenant_contacts WHERE tenant_id =:tenantId AND cw_id =:cwId")
                    .as(Long.class)
                    .bind("tenantId", contacts.getTenantId())
                    .bind("cwId", contacts.getCwId())
                    .fetch().one().blockOptional(Duration.ofSeconds(10));
            if (id.isPresent() && id.get() > 0) {
                log.debug("Receiver connect wise tenant id:{},cw id:{},contacts update id:{}",
                        contacts.getTenantId(), contacts.getCwId(), contacts.getId());
                contacts.setId(id.get());
                contacts.setSyncTime(LocalDateTime.now());
                this.databaseClient.update().table(Contacts.class).using(contacts).fetch().rowsUpdated().block();
            } else {
                this.databaseClient.insert().into(Contacts.class).using(contacts).fetch().rowsUpdated().block();
            }
//        contacts.setExtend(null);
//        super.messagingTemplate.convertAndSend("next-ticket.redis.contacts", contacts);

        }catch (Exception e){
            if (!ObjectUtils.isEmpty(contact.get("tenantId"))){
                this.databaseClient.insert().into(ExceptionLog.class).using(
                        ExceptionLog.builder()
                                .tenantId(contact.get("tenantId").asInt())
                                .name("wise.contacts")
                                .status(500)
                                .type("wise.contacts")
                                .msg(e.getMessage()+","+contact.toString() )
                                .build()
                ).fetch().rowsUpdated().block(Duration.ofSeconds(10));
            }
        }
    }

}
